package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author sw
 * @CreateTime 2024-11-17
 */
public class DwdTradeOrderRefund extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderRefund().start(Constant.TOPIC_DWD_TRADE_ORDER_REFUND,4,10017);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        //1、读取ods
        readOdsDb(tableEnv,groupId);
        //2、读取字段表
        createBaseDic(tableEnv);
        //3、过滤退单表数据
        getOrderRefundInfoTable(tableEnv);
        //4、过滤订单表中的退单数据
        getOrderIndoTable(tableEnv);
        //5、join
        Table result = getResult(tableEnv);
        //6、写出到kafka
        tableEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_ORDER_REFUND + "(" +
                "id string," +
                "user_id string," +
                "order_id string," +
                "sku_id string," +
                "province_id string," +
                "date_id string," +
                "create_time string," +
                "refund_type_code string," +
                "refund_type_name string," +
                "refund_reason_type_code string," +
                "refund_reason_type_name string," +
                "refund_reason_txt string," +
                "refund_num string," +
                "refund_amount string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
        result.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_REFUND).execute();

    }

    private static Table getResult(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery(" select\n" +
                " ri.id,\n" +
                " ri.user_id,\n" +
                " ri.order_id,\n" +
                " ri.sku_id,\n" +
                " oi.province_id,\n" +
                " date_format(ri.create_time,'yyyy-MM-dd') date_id,\n" +
                " ri.create_time,\n" +
                " ri.refund_type,\n" +
                " b.info.dic_name,\n" +
                " ri.refund_reason_type,\n" +
                " c.info.dic_name,\n" +
                " ri.refund_reason_txt,\n" +
                " ri.refund_num,\n" +
                " ri.refund_amount,\n" +
                " ri.ts\n" +
                " from order_refund_info ri\n" +
                " join order_info oi\n" +
                " on ri.order_id = oi.id\n" +
                " JOIN base_dic FOR SYSTEM_TIME AS OF ri.proc_time AS b\n" +
                " ON ri.refund_type = b.rowkey\n" +
                " JOIN base_dic FOR SYSTEM_TIME AS OF ri.proc_time AS c\n" +
                " ON ri.refund_reason_type = c.rowkey");
    }

    /**
     * 4、过滤订单表中的退单数据
     * @param tableEnv
     */
    private static void getOrderIndoTable(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(" select\n" +
                " `data`['id'] id,\n" +
                " `data`['province_id'] province_id,\n" +
                " `old`\n" +
                " from topic_db \n" +
                " where `database` = 'gmall_realtime' \n" +
                " and `table` = 'order_info' \n" +
                " and `type` = 'update' \n" +
                " and `old`['order_status'] is not null \n" +
                " and `data`['order_status'] = '1005'");
        tableEnv.createTemporaryView("order_info",table);
    }

    /**
     * 3、过滤退单表数据
     * @param tableEnv
     */
    private static void getOrderRefundInfoTable(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(" select\n" +
                " `data`['id'] id,\n" +
                " `data`['user_id'] user_id,\n" +
                " `data`['order_id'] order_id,\n" +
                " `data`['sku_id'] sku_id,\n" +
                " `data`['refund_type'] refund_type,\n" +
                " `data`['refund_num'] refund_num,\n" +
                " `data`['refund_amount'] refund_amount,\n" +
                " `data`['refund_reason_type'] refund_reason_type,\n" +
                " `data`['refund_reason_txt'] refund_reason_txt,\n" +
                " `data`['create_time'] create_time,\n" +
                " proc_time,\n" +
                " ts\n" +
                " from topic_db\n" +
                " where `database` = 'gmall_realtime'\n" +
                " and `table` = 'order_refund_info'\n" +
                " and `type` = 'insert'");
        tableEnv.createTemporaryView("order_refund_info",table);
    }
}
